  
  package com.zs.pig.storm;  
import java.io.File;
import java.io.IOException;
import java.util.Arrays;
import java.util.Map;

import net.oschina.common.search.IndexHolder;

import org.apache.commons.io.FileUtils;

import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import et.oschina.common.search.test.Person;
import et.oschina.common.search.test.Post;
import et.oschina.common.search.test.SnowflakeIdWorker;
  
/** 
 * Created by IntelliJ IDEA. User: comaple.zhang Date: 12-8-28 Time: 下午2:11 To 
 * change this template use File | Settings | File Templates. 
 */  
@SuppressWarnings("serial")  
public class AllDatatoLuenceBolt extends BaseRichBolt {  
  
    private OutputCollector collector;  
    IndexHolder holder;
    LogBean logBean=null;
    Post post1 = new Post(1, "post title1", "post body1");
    Post post2 = new Post(2, "阿里云携手开源中国众包平台发布百万悬赏项目", "阿里云与开源中国达成战略合作，首期将通过开源中国众包平台发布近百万元悬赏项目");
    Post post3 = new Post(3, "SQLite 3.9.0 发布，数据库服务器", "SQLite是遵守ACID的关联式数据库管理系统，它包含在一个相对小的C库中。它是D.RichardHipp建立的公有领域项目。");
    @Override  
    public void prepare(Map stormConf, TopologyContext context,  
            OutputCollector collector) {  
        this.collector = collector;  
        String indexFolder = AllDatatoLuenceBolt.class.getClassLoader().getResource("./").getFile() + "index";
       
        try {
        	System.out.println("filepath:"+indexFolder);
			FileUtils.deleteDirectory(new File(indexFolder));
			 FileUtils.forceMkdir(new File(indexFolder));
		        holder = IndexHolder.init(indexFolder);
		        
		       
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
       
    }  
  
    @Override  
    public void execute(Tuple input) {  
        try {  
        	SnowflakeIdWorker idWorker = new SnowflakeIdWorker(0, 0);
        	String mesg = input.getString(0);  
                String [] msgs=mesg.split(" - ");
            //    if (mesg != null && msgs.length == 9) {  
                	System.out.println("luence");
                	logBean=new LogBean(idWorker.nextId(),mesg.split(" - ")[3],mesg.split(" - ")[1],mesg.split(" - ")[2],mesg.split(" - ")[0]);
                	holder.add(Arrays.asList(logBean));
                    for (int i = 0; i < 1; i++)
                        holder.optimize(LogBean.class);
                    
//                	JDBCUtil.update(INSERT_LOG, mesg.split(" - ")[0],mesg.split(" - ")[1],mesg.split(" - ")[2],mesg.split(" - ")[3],
//                             mesg.split(" - ")[4],mesg.split(" - ")[5],mesg.split(" - ")[6],mesg.split(" - ")[7],mesg.split(" - ")[8]); 
             //   }
  
        } catch (Exception e) {  
            e.printStackTrace(); // To change body of catch statement use File |  
            collector.fail(input);                  // Settings | File Templates.  
        }  
        collector.ack(input);  
    }  
  
    @Override  
    public void declareOutputFields(OutputFieldsDeclarer declarer) {  
        declarer.declare(new Fields("id","data"));  
    }  
}  